package com.doit.spark.day05

import com.alibaba.fastjson.JSON
import com.doit.spark.day01.utils.SparkUtil
import org.apache.spark.util.LongAccumulator

/**
 * @DATE 2022/1/8/9:35
 * @Author MDK
 * @Version 2021.2.2
 *
 *   过滤脏数据
 * */
object C01_累加器 {
  def main(args: Array[String]): Unit = {
    val sc = SparkUtil.getSc
    val rdd = sc.textFile("data/movie/")

    //使用累加器
    val err_cnt: LongAccumulator = sc.longAccumulator("err_cnt")
    val succ_cnt: LongAccumulator = sc.longAccumulator("succ_cnt")
    val resRDD = rdd.map(line => {
      var bean: MovieBean = null
      //处理行数据  解析数据
      /* classOf[MovieBean]
      *   MovieBean.class
      *   Class.forName("类名")
      * */
      try {
        val bean = JSON.parseObject(line, classOf[MovieBean])
        succ_cnt.add(1)
      } catch {
        case e: Exception => {
          err_cnt.add(1)
          println(s"$line 出现异常了...")
        }
      }
      bean
    }).filter(_ != null)
    //行动算子
    resRDD.foreach(println)
    //Driver端聚合打印结果
    println("错误行数"+err_cnt.value+"行")
    println("正确数据"+succ_cnt.value+"行")

  }
}
